package com.flink.demo.wordcount04;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * Windows使用 nc.exe -L -v -p 9100监听端口
 *
 * @author zsyoung@qq.com
 * 2020/7/4 1:04
 */
public class WordCountDataStream {
    public static void main(String[] args) throws Exception {
        //1、构建流式环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2、监听本地9000端口
        DataStream<String> text = env.socketTextStream("127.0.0.1", 9100, "\n");
        //3、将接收的数据进行拆分，分组，窗口计算并且进行聚合输出
        DataStream<WordWithCount> windowCounts = text.flatMap(new FlatMapFunction<String, WordWithCount>() {
            @Override
            public void flatMap(String value, Collector<WordWithCount> collector) throws Exception {
                for (String word : value.split("\\s")) {
                    collector.collect(new WordWithCount(word, 1L));
                }
            }
        })
                .keyBy("word")
                //每1秒计算一次最近5秒的数据
                .timeWindow(Time.seconds(5), Time.seconds(1))
                .reduce(new ReduceFunction<WordWithCount>() {
                    @Override
                    public WordWithCount reduce(WordWithCount a, WordWithCount b) throws Exception {
                        return new WordWithCount(a.word, a.count + b.count);
                    }
                });
//                .reduce((ReduceFunction<WordWithCount>) (a, b) -> new WordWithCount(a.word, a.count + b.count));

        //4、打印结果
        windowCounts.print().setParallelism(1);

        env.execute("Socket_Window_Wordcount");
    }

    public static class WordWithCount {
        public String word;
        public long count;

        public WordWithCount() {
        }

        public WordWithCount(String word, long count) {
            this.word = word;
            this.count = count;
        }

        @Override
        public String toString() {
            return word + ":" + count;
        }
    }
}
